package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Configuration
@Slf4j
public class HotArticleStreamHandler {

    /**
     * 构建流处理程序
     * @param streamsBuilder
     * @return
     */
    @Bean
    public KStream<String,String> kStream(StreamsBuilder streamsBuilder){
        //接收消息(来自ApLikesBehaviorServiceImpl里的：
        //1.ApLikesBehaviorServiceImpl发送点赞的消息(实时更新文章的点赞数) stream数据聚合(需要先注入kafuka)
//        kafkaTemplate.send(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC,
//            JSON.toJSONString(mess));
        //发消息给kafka，kafka再发消息给stream，让它聚合数据，然后更新聚合后的数据到redis

        //2.ApReadBehaviorServiceImpl发送消息(实时更新文章的阅读数) stream数据聚合(需要先注入kafuka)
//    UpdateArticleMess mess = new UpdateArticleMess();
//        mess.setArticleId(dto.getArticleId());
//        mess.setType(UpdateArticleMess.UpdateArticleType.VIEWS);
//        mess.setAdd(1);
//        kafkaTemplate.send(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC,
//            JSON.toJSONString(mess));

        KStream<String,String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);
        //评论/行为微服务的kafkaTemplate.send(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC,JSON.toJSONString(mess));
        // 发消息给kafka，kafka再发消息给stream，让它聚合数据，然后更新聚合后的数据到redis？？？？
        //聚合流式处理
        stream.map((key,value)->{
            //在接收到前把mess转换成了JSON(JSON.toJSONString(mess))，这里要将消息的value转换回对象
            UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
            //重置消息的key:1234343434   和  value: likes:1
            //因为要根据当前的文章id进行聚合(计算某个文章id有多少个点赞/阅读数)，所以要把当前的这个key设置为文章的id
            //因为泛型都是string，所以要把id转成一个字符串
            //TODO 当前文章点赞了一次，重置当前的消息的key和value????
            //                                                key,value
            //               (就是下面apply(String key, String value, String aggValue)里的key和value)
            //                                                                 valAry[0]:valAry[1]
            return new KeyValue<>(mess.getArticleId().toString(),mess.getType().name()+":"+mess.getAdd());
        })
                //按照当前的(key)文章id进行聚合
                .groupBy((key,value)->key)
                //时间窗口，10秒聚合一次
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                /**
                 * 自行的完成聚合的计算
                 */
                .aggregate(new Initializer<String>() {
                    /**
                     * 初始方法，返回值是消息的value
                     * @return
                     */
                    @Override //初始组合的数据
                    public String apply() {
                        //        split[0]:split[1]
                        return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                    }
                    /**
                     * 真正的聚合操作，返回值是处理之后的消息的value
                     */
                }, new Aggregator<String, String, String>() {
                    /**
                     * @param key 消息的key(文章id)
                     * @param value 消息的value 当前的某一个类型 likes:1 (上面的mess.getType().name()+":"+mess.getAdd())
                     * @param aggValue 上面apply()的返回值： COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
                     * @return
                     */
                    @Override
                    public String apply(String key, String value, String aggValue) {
                        System.out.println("第一个apply()里的value是："+value);
                        if(StringUtils.isBlank(value)){
                            log.info("value为空，直接返回初始值apply()的返回值：", aggValue);
                            return aggValue;
                        }
                        //"COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0"
                        String[] aggAry = aggValue.split(",");
                        log.info("aggValue.split是{}",aggAry);
                        int col = 0,com=0,lik=0,vie=0;
                        for (String agg : aggAry) {
                            String[] split = agg.split(":");
                            log.info("agg.split是{}",split);
                            /**
                             * 获得初始值(=是赋值)，也是时间窗口内(10秒)计算之后的值
                             */
                            //split[0]就是操作的类型是"COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0"
                            // 里面的COLLECTION/COMMENT/LIKES/VIEWS
                            //split[1]就是初始值 0 是"COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0"
                            //里面的某一个0 或者是 当前时间窗口(10秒)内文章该操作类型的总数
                            log.info("split[0]：{}和split[1]：{}",split[0],split[1]);
                            //匹配上哪个类型，就给对应的值赋值
                            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                                case COLLECTION:
                                    //split[1]是当前时间窗口(10秒)内文章该操作类型的总数(初始值是0)
                                    col = Integer.parseInt(split[1]);
                                    break;
                                case COMMENT:
                                    com = Integer.parseInt(split[1]);
                                    break;
                                case LIKES:
                                    lik = Integer.parseInt(split[1]);
                                    break;
                                case VIEWS:
                                    vie = Integer.parseInt(split[1]);
                                    break;
                            }
                        }
                        /**
                         * 累加(+=是累加)操作   likes:1
                         */
                        //因为上面new KeyValue<>(mess.getArticleId().toString(),mess.getType().name()+":"+mess.getAdd());
                        //组装的时候，就是用 : 来组装，所以这里也是用 : 来拆分
                        String[] valAry = value.split(":");
                        //valAry[0]是操作的类型(mess.getType().name())是COLLECTION/COMMENT/LIKES/VIEWS
                        //valAry[1]是mess.getAdd() 是 点赞(+1) 或者 取消点赞(-1) 或者 浏览一次(+1)...
                        switch (UpdateArticleMess.UpdateArticleType.valueOf(valAry[0])){
                            case COLLECTION:
                                //valAry[1]就是mess.getAdd()
                                col += Integer.parseInt(valAry[1]);
                                break;
                            case COMMENT:
                                com += Integer.parseInt(valAry[1]);
                                break;
                            case LIKES:
                                lik += Integer.parseInt(valAry[1]);
                                break;
                            case VIEWS:
                                vie += Integer.parseInt(valAry[1]);
                                break;
                        }

                        //如果想让apply()里的return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";存值，需要返回相同格式的字符串
                        String formatStr = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", col, com, lik, vie);
                        log.info("文章的id：{}",key);
                        log.info("当前时间窗口内的消息处理结果：{}",formatStr);
                        return formatStr;
                    }
                    //聚合之后的数据，存储在redis中，key是文章id，value是文章的统计数据
                    //hot-atricle-stream-count-001是当前流式处理的一个状态，如果后期有多个流式处理，只要不重复就行
                }, Materialized.as("hot-atricle-stream-count-001"))
                .toStream()
                .map((key,value)->{
                    //不知道是哪一个文章，所以需要传递文章id
                    //value就是formatStr(里面没有文章id，所以需要传递文章id，所以需要专门处理，在下面定义formatObj()方法)
                    //                           文章id        ,      文章id                  ,消息的值
                    return new KeyValue<>(key.key().toString(),formatObj(key.key().toString(),value));
                })
                //发送消息(内容就是formatObj(key.key().toString(),value)返回的内容)
                //发给listener.ArticleIncrHandleListener里的@KafkaListener(topics = HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC)
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);
        return stream;
    }

    /**
     * 格式化消息的value数据
     * @param articleId 当前某个文章的id
     * @param value COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
     * @return
     */
    public String formatObj(String articleId,String value){
        log.info("格式化消息的value数据，文章的id：{}，value：{}",articleId,value);
        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();

        mess.setArticleId(Long.valueOf(articleId));
        //COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
        String[] valAry = value.split(",");
        for (String val : valAry) {
            String[] split = val.split(":");
            //split[0]就是"COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0"里的某一个类型
            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                case COLLECTION:
                    mess.setCollect(Integer.parseInt(split[1]));
                    break;
                case COMMENT:
                    mess.setComment(Integer.parseInt(split[1]));
                    break;
                case LIKES:
                    mess.setLike(Integer.parseInt(split[1]));
                    break;
                case VIEWS:
                    mess.setView(Integer.parseInt(split[1]));
                    break;
            }
        }
        log.info("格式化消息，聚合消息处理之后的结果为:{}",JSON.toJSONString(mess));
        return JSON.toJSONString(mess);

    }
}